insert overwrite table jms_dwd.dwd_order_source_detail_dt partition (dt)
select
     billcode
    ,ordersource_code
    ,ordersource_name
    ,input_time
    ,subscribe_point
    ,dt
from (
    select
          way.waybill_no            as billcode
         ,way.subscribe_source_code as ordersource_code
         ,way.subscribe_source_name as ordersource_name
         ,way.input_time            as input_time
         ,if(way.subscribe_source_code is not null, '1', '0') as subscribe_point
         ,way.dt
    from (
         select
               waybill_no
              ,input_time
              ,subscribe_source_code
              ,subscribe_source_name
              ,'{{ execution_date | cst_ds }}' as dt
         from jms_ods.yl_oms_oms_waybill--运单表
         where dt = '{{ execution_date | cst_ds }}'
           and to_date(input_time) = '{{ execution_date | cst_ds }}'
    ) way
    union all
    select
          der.billcode              as billcode
         ,way.subscribe_source_code as ordersource_code
         ,way.subscribe_source_name as ordersource_name
         ,der.input_time            as input_time
         ,if(way.subscribe_source_code is not null, '1', '0') as subscribe_point
         ,der.dt
    from jms_dwd.dwd_order_source_detail_dt der
    left join (
       select
             waybill_no
            ,input_time
            ,subscribe_source_code
            ,subscribe_source_name
            ,'{{ execution_date | cst_ds }}' as dt
       from jms_ods.yl_oms_oms_waybill--运单表
       where dt = '{{ execution_date | cst_ds }}'
    ) way on der.billcode = way.waybill_no
    where der.dt >= date_add('{{ execution_date | cst_ds }}', -2)
      and der.dt < '{{ execution_date | cst_ds }}'
      and subscribe_point = '0'
    union all
    select
          billcode
         ,ordersource_code
         ,ordersource_name
         ,input_time
         ,subscribe_point
         ,dt
    from jms_dwd.dwd_order_source_detail_dt --前两天订阅的历史数据不需要回刷
    where dt >= date_add('{{ execution_date | cst_ds }}', -2)
      and dt < '{{ execution_date | cst_ds }}'
      and subscribe_point = '1'
) a
distribute by abs(hash(billcode)) % 24;